package hanlen.mq.starter.support;

import hanlen.mq.starter.annotation.RedisQueueListener;
import hanlen.mq.starter.bean.MessageWrap;
import hanlen.mq.starter.configuration.properties.HanlenMqProperties;
import hanlen.mq.starter.constants.QueueConstants;
import hanlen.mq.starter.core.RedisQueueExportService;
import hanlen.mq.starter.core.RedissonService;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RAtomicLong;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

/**
 * archives-service
 *
 * @author MrYe
 * @Description
 * @date 2021-10-23 16:38
 */
@Slf4j
public class RedisQueueDlqSupport implements ApplicationContextAware {

    @Autowired
    private RedisQueueExportService exportService;
    @Autowired
    private RedissonService redissonService;
    @Autowired
    private HanlenMqProperties mqProperties;
    private ApplicationContext context;

    public void monitor() {
        final Map<String, RedisQueueListener> listenerMap = context.getBeansOfType(RedisQueueListener.class);
        listenerMap.forEach((beanName, listener) -> {
            final String dlqName = QueueConstants.getDeadQueueKey(listener.name());
            final long msgSize = redissonService.getAtomicLong(dlqName).get();
            // 队列存在
            if (msgSize > 0) {
                exportService.alarm(dlqName, (int) msgSize);
            }
        });
    }

    public void addToDeadQueue(RedisQueueListener annotation, MessageWrap message) {
        final String queueName = annotation.name();
        final String dlqName = StringUtils.endsWith(queueName, QueueConstants.DLQ_SUFFIX) ? queueName : queueName + QueueConstants.DLQ_SUFFIX;
        try {
            if (!annotation.retry()) {
                exportService.overflowPolicy(queueName, message);
                return;
            }
            // 同一个消息可以重复转存处理3次
            final String retryKey = QueueConstants.getRetryKey(message.getMessageId());
            final RAtomicLong atomicLong = redissonService.getAtomicLong(retryKey);
            final int intervalSize = mqProperties.getDlqRetryInterval().size();

            if (atomicLong.get() >= intervalSize) {
                redissonService.delete(retryKey);
                log.error("hanlen-redis-queue:{} retry:{} times failed", queueName, atomicLong.get());
                exportService.overflowPolicy(queueName, message);
                return;
            }
            final Integer[] retryIntervals = mqProperties.getDlqRetryInterval().toArray(new Integer[intervalSize]);
            final long retryTimes = atomicLong.incrementAndGet();

            final long delay = retryTimes * retryIntervals[(int) retryTimes - 1];
            redissonService.addDelayedQueue(dlqName, new HashMap<>(0), message.getBody(), delay, TimeUnit.SECONDS);
            log.info("hanlen-redis-queue:{} retry:{}times success block message-size:{} delay:{}s executor", dlqName, retryTimes,
                redissonService.getAtomicLong(QueueConstants.getSizeKey(dlqName)).get(), delay);
        } catch (Exception e) {
            log.error("hanlen-redis-queue:{} retry failed", dlqName, e);
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = applicationContext;
    }

}
